package com.wang.lottery.application.mp.consumer;

import cn.hutool.core.lang.Assert;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.wang.lottery.application.mp.producer.KafkaProducer;
import com.wang.lottery.common.Constants;
import com.wang.lottery.domain.activity.model.vo.InvoiceVO;
import com.wang.lottery.domain.award.model.req.GoodsReq;
import com.wang.lottery.domain.award.model.res.DistributionRes;
import com.wang.lottery.domain.award.service.factory.DistributionGoodsFactory;
import com.wang.lottery.domain.award.service.goods.IDistributionGoods;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Optional;

/**
 * @author wjh
 * @create 2023-11-12 17:11
 * 消息消费者
 */
@Component
@Slf4j
public class LotteryInvoiceListener {

    @Resource
    private DistributionGoodsFactory distributionGoodsFactory;

    @KafkaListener(topics = KafkaProducer.TOPIC_LOTTERY_RESULT, groupId = KafkaProducer.TOPIC_LOTTERY_GROUP)
    public void listen(ConsumerRecord<?, ?> message, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        Optional<?> optional = Optional.ofNullable(message.value());
        if (!optional.isPresent()) {
            return;
        }

        try {
            Object msg = optional.get();
            // 1. 转化对象（或者你也可以重写Serializer<T>）
            InvoiceVO invoiceVO = JSONUtil.toBean(JSONUtil.parseObj(msg), InvoiceVO.class);
            // 2. 获取发送奖品工厂，执行发奖
            IDistributionGoods goodsService = distributionGoodsFactory.getDistributionGoodsService(invoiceVO.getAwardType());
            DistributionRes distributionRes = goodsService.doDistribution(new GoodsReq(
                    invoiceVO.getuId(),
                    invoiceVO.getOrderId().toString(),
                    invoiceVO.getAwardId(),
                    invoiceVO.getAwardName(),
                    invoiceVO.getAwardContent()));
            Assert.isTrue(distributionRes.getCode().equals(Constants.AwardState.SUCCESS.getCode()));

            log.info("消费MQ消息，完成 topic：{} uId：{} 发奖结果：{}", topic, invoiceVO.getuId(), JSON.toJSONString(distributionRes));

            ack.acknowledge();
        } catch (Exception e) {
            log.error("消费MQ消息，发生异常，topic：{} uId：{} 异常信息：{}", topic, message.key(), e.getMessage());
            throw e;
        }
    }
    
    
}
